package com.beesphere.products.throttling.processor.async;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.CancellationException;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.beesphere.products.throttling.AsyncCallback;
import com.beesphere.products.throttling.EnhancedThread;
import com.beesphere.products.throttling.UnitOfWork;

public abstract class ThreadProcessor implements AsyncProcessor {

	private static final Logger logger = LoggerFactory.getLogger (ThreadProcessor.class);
	
	private ExecutorService executor;
    private final AtomicBoolean shutdown = new AtomicBoolean (true);
    private boolean callerRunsWhenRejected = true;

    public void process (UnitOfWork unitOfWork) throws Exception {
        AsyncProcessorHelper.process (this, unitOfWork);
    }

    public boolean process (final UnitOfWork unitOfWork, final long timeout, final AsyncCallback callback) {
        if (shutdown.get()) {
            throw new IllegalStateException ("ThreadProcessor is not running.");
        }
        List<Callable<Object>> toRun = new ArrayList<Callable<Object>>();
		toRun.add (new Callable<Object> () {
			public Object call() throws Exception {
	            if (shutdown.get ()) {
	            	unitOfWork.setError (new RejectedExecutionException ("Throttler is in shutdown state"));
	            }
	            
	            if (unitOfWork.getError () != null) {
		            return null;
	            }
	            
	            // execute the unitOfWork and call the callback
	            logger.debug ("Execute: [" + Thread.currentThread ().getThreadGroup().getName () + "] - " 
	            		+ Thread.currentThread ().getName () + ", TimeOut[" + timeout + "], UOW ID: [" + unitOfWork.getId () + "]");
	            
	            EnhancedThread.setCurrentContext (unitOfWork.getContext ());
	            
	            Object result = unitOfWork.execute ();
	            
	            // call the callback
	            logger.debug ("\tCallback: [" + Thread.currentThread ().getName () + "]");
	            
	            unitOfWork.setFinished (true);
	            callback.done (result, false);
	            
	            return result;
			}
		});
		
		List<Future<Object>> futures = null;
		
        try {
        	if (timeout <= 0) {
            	futures = executor.invokeAll (toRun);
        	} else {
            	futures = executor.invokeAll (toRun, timeout, TimeUnit.MILLISECONDS);
        	}
        	try {
    			// Find out what happened when the service was called.
    			futures.get (0).get ();
	   		} catch (ExecutionException e) {
    			// Propagate the exception that is part of the
    			// interface.
    			unitOfWork.setError (e.getCause ());
    			// call the callback
	            logger.error ("\tCallback-OnError: [" + Thread.currentThread ().getName () + "], Error: ", e.getMessage ());
	            logger.info ("", e);
	            callback.done (null, false);
	            unitOfWork.setFinished (true);
    		} catch (CancellationException e) {
    			unitOfWork.setCanceled (true);
    			unitOfWork.setError (e);
    			logger.error ("\tCallback-OnCancel: [" + Thread.currentThread ().getName () + "], Error: ", e.getMessage ());
	            logger.info ("", e);
    			callback.done (null, false);
    			unitOfWork.setFinished (true);
    		}
    		return false;
        } catch (InterruptedException e) {
        	if (callerRunsWhenRejected) {
                if (shutdown.get ()) {
                	unitOfWork.setError (new RejectedExecutionException ("Throttler is in shutdown state"));
                } 
            } else {
            	unitOfWork.setError (e);
            }
        	logger.error ("\tCallback-OnReject: [" + Thread.currentThread ().getName () + "], Error: ", unitOfWork.getError ());
        	callback.done (null, true);
        	unitOfWork.setFinished (true);
            return true;
        }
        
    }

    public void start () throws Exception {
        shutdown.set (false);
        executor = createExecutor ();
    }
    
    public void stop () throws Exception {
        shutdown.set (true);
        executor.shutdown ();
        executor.awaitTermination (0, TimeUnit.SECONDS);
    }

    public boolean isCallerRunsWhenRejected() {
        return callerRunsWhenRejected;
    }

    public void setCallerRunsWhenRejected(boolean callerRunsWhenRejected) {
        this.callerRunsWhenRejected = callerRunsWhenRejected;
    }

    protected abstract ExecutorService createExecutor ();

}
